package com.sunzm.flink.loginfail

import java.time.Duration
import java.util

import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.DateUtils
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.cep.functions.{PatternProcessFunction, TimedOutPartialMatchHandler}
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}


/**
 * 使用CEP编程对2秒内连续登录失败3次的用户进行告警
 */
object CepLoginFailCheckJob {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  val failTimeout = new OutputTag[String]("failTimeout")

  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    //为了便于观察，并行度设置为1，实际生产环境不能设置成1
    env.setParallelism(1)

    //读取数据
    //用户id,时间,类型（0成功,1失败）
    //u1001,20210704175816001,0
    val sourceDS: DataStream[String] = env.fromElements[String](
      "u1001,20210704175816001,1",
      "u1001,20210704175816002,1",
      "u1001,20210704175816003,1",
      "u1001,20210704175816004,1",
      "u1001,20210704175816005,1",
      "u1001,20210704175816006,1"
    )

    //val sourceDS = env.socketTextStream("82.156.210.70", 9999)

    //CEP需要使用事件时间语义
    val watermarkStrategy = WatermarkStrategy
      //数据没有乱序，WaterMark不需要设置乱序程度
      .forBoundedOutOfOrderness[UserAction](Duration.ofSeconds(0))
      .withTimestampAssigner(new SerializableTimestampAssigner[UserAction] {
        override def extractTimestamp(event: UserAction, recordTimestamp: Long): Long = {
          val eventTime: Long = DateUtils.parseDate(event.dateTimeStr, "yyyyMMddHHmmssSSS").getTime
          eventTime
        }
      })

    val dataStream: DataStream[UserAction] = sourceDS
      .filter(!_.isEmpty)
      .map(line => {
        val fields = StringUtils.split(line, ",")
        UserAction(fields(0), fields(1), fields(2).toInt)
      }).assignTimestampsAndWatermarks(watermarkStrategy)

    val keyedStream: KeyedStream[UserAction, String] = dataStream.keyBy(_.userId)

    //定义一个模式
    val pattern = Pattern.begin[UserAction]("begin")
      .where(_.actionType == 1)
      .times(3)
      .within(Time.seconds(2))

    //应用模式
    val patternStream = CEP.pattern(keyedStream, pattern)

     val resultDS: DataStream[String] = patternStream.process(new MyPatternProcessFunction)

     resultDS.print("主流")

     //获取侧输出流中的数据
     resultDS.getSideOutput(failTimeout).print("侧流")

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }

  private class MyPatternProcessFunction extends PatternProcessFunction[UserAction, String]
    with TimedOutPartialMatchHandler[UserAction] {
    /**
     * 处理匹配到的数据
     *
     * @param matchPattenMap
     * @param ctx
     * @param out
     */
    override def processMatch(matchPattenMap: util.Map[String, util.List[UserAction]],
                              ctx: PatternProcessFunction.Context,
                              out: Collector[String]): Unit = {

      val userActions: util.List[UserAction] = matchPattenMap.get("begin")

      val userId = userActions.get(0)

      val outBuilder = new StringBuilder

      outBuilder.append("=======================================================").append(System.lineSeparator())

      outBuilder.append(s"${userId}, 2秒内登录失败次数超过3次, 明细如下:").append(System.lineSeparator())


      import scala.collection.JavaConversions._

      userActions.foreach(action => {
        outBuilder.append(s"${action.dateTimeStr}, ${action.actionType}").append(System.lineSeparator())
      })

      outBuilder.append("=======================================================").append(System.lineSeparator())

      out.collect(outBuilder.toString())
    }

    /**
     * 处理超时的数据
     *
     * @param timeOutMap
     * @param ctx
     */
    override def processTimedOutMatch(timeOutMap: util.Map[String, util.List[UserAction]],
                                      ctx: PatternProcessFunction.Context): Unit = {

      val userAction = timeOutMap.get("begin").get(0)
      //输出到侧输出流
      ctx.output(failTimeout, s"${userAction.userId}, 2秒内登录失败次数不足 3 次, 重新进行统计...")
    }
  }

  private case class UserAction(
                                 /**
                                  * 用户ID
                                  */
                                 userId: String,

                                 /**
                                  * 操作时间 yyyyMMddHHmmssSSS
                                  */
                                 dateTimeStr: String,

                                 /**
                                  * 操作类型 0成功，1失败
                                  */
                                 actionType: Int)

}
